package com.example.actordemo3.total;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.Terminated;
import akka.cluster.routing.ClusterRouterPool;
import akka.cluster.routing.ClusterRouterPoolSettings;
import akka.routing.RoundRobinPool;
import akka.routing.Router;

import java.util.ArrayList;
import java.util.List;

public class WordFrontService extends AbstractActor {

    List<ActorRef> wordCountServices = new ArrayList<>();

    private  int jobCounter = 0;

    ActorRef router = null;

    @Override
    public void preStart() throws Exception {
        int totalInstance = 100;
        int maxnode = 5;
        boolean allowLocalRoutees =false;
        String userRole = null;
        router  =  getContext().actorOf(new ClusterRouterPool(new RoundRobinPool(3),
                new ClusterRouterPoolSettings(totalInstance,maxnode,allowLocalRoutees,userRole))
                .props(Props.create(WordCountService.class)),"router");
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder().match(Article.class ,msg->{
//            jobCounter++;
            Article article = msg;
//            int index = jobCounter % wordCountServices.size();
//
//            System.out.println("选着节点"+index);
//            wordCountServices.get(index).forward(article,getContext());
            router.tell(article,getSender());
        }).match(String.class,msg->{
            if(msg.equals("serviceIsOk")){
                ActorRef sender = getSender();
                System.out.println(sender+"可用");
                wordCountServices.add(sender);
                getContext().watch(sender);
            }else  if(msg.equals("isready")){
                if(!wordCountServices.isEmpty()){
                    getSender().tell("ready",getSelf());
                }else {
                    getSender().tell("notReady",getSelf());
                }

            }
        }).match(Terminated.class,msg->{
            Terminated t = msg;
            System.out.println("移除"+t.getActor());
            wordCountServices.remove(t.getActor());
        }).matchAny(o->{
            unhandled(o);
        }).build();
    }
}
